package com.googlecode.etl.kravas;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.*;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.util.Assert;

import java.util.List;

/**
 * @author A_Reshetnikov
 * @since Date: 28.08.2013
 */
public class JobRunner {
    protected static final Log logger = LogFactory.getLog(JobRunner.class);
    private JobLauncher launcher;

    int start(ConfigurableApplicationContext context, String inputFileName, String outputFileName) throws Exception {
        try {
            context.refresh();
            context.getAutowireCapableBeanFactory().autowireBeanProperties(this,
                    AutowireCapableBeanFactory.AUTOWIRE_BY_TYPE, false);

            Assert.state(launcher != null, "A JobLauncher must be provided.  Please add one to the configuration.");
            Job job = (Job) context.getBean("job1");


            JobParametersBuilder parametersBuilder = new JobParametersBuilder();
            parametersBuilder.addString("inputFileName", "file:" + inputFileName);
            parametersBuilder.addString("outputFileName", outputFileName);
            JobParameters jobParameters = parametersBuilder.toJobParameters();
            logger.info(String.format("Start job with parameters %s", jobParameters));
            JobExecution jobExecution = launcher.run(job, jobParameters);
            //write some log
            writeJobExecutionStatus(jobExecution, jobParameters);
            if (jobExecution.getStatus().isUnsuccessful()) {
                throw new Exception(String.format("Job %s unsuccessful", jobParameters));
            }
            //    logger.info(String.format("Processing schema %s %s completed ", schemaList.size() > 1 ? "list" : "", schemaList));

            return 1;

        } catch (Exception e) {
            String message = "Job Terminated in error: " + e.getMessage();
            logger.error(message, e);
            throw e;
        } finally {
            if (context != null) {
                context.close();
            }
        }
    }

    private JobParameters getJobParameters(String schemaName, boolean launchedByDBA) {
        JobParametersBuilder parametersBuilder = new JobParametersBuilder();
        return parametersBuilder.toJobParameters();
    }

    private void writeJobExecutionStatus(JobExecution jobExecution, JobParameters jobParameters) {
        StepExecution step = jobExecution.getStepExecutions().toArray(new StepExecution[]{})[0];
//        logger.info(String.format("Processed %d items from total %d in file %s",
//                step.getWriteCount(), step.getReadCount(), jobParameters.getString("inputFileName")));
        logger.info(String.format("Skip processing %d items from total %d in file %s",
                step.getSkipCount(), step.getReadCount(), jobParameters.getString("inputFileName")));
        long seconds = ((step.getEndTime().getTime() - step.getStartTime().getTime()) / 1000);
        logger.info(String.format("Job is %s in %d seconds", step.getExitStatus().getExitCode(), seconds));
    }

    public void setLauncher(JobLauncher launcher) {
        this.launcher = launcher;
    }


}
